package com.shanks.framework.mq.service.impl;

import cn.hutool.core.util.IdUtil;
import com.shanks.framework.mq.common.constant.MQConstant;
import com.shanks.framework.mq.common.util.LogUtil;
import com.shanks.framework.mq.domain.MQMessage;
import com.shanks.framework.mq.enums.MQZone;
import com.shanks.framework.mq.exception.MQDuplicatedConsumeException;
import com.shanks.framework.mq.model.MessageConsumeExceptionRecordPO;
import com.shanks.framework.mq.model.MessageConsumeRecordPO;
import com.shanks.framework.mq.repository.MessageConsumeExceptionRecordRepository;
import com.shanks.framework.mq.repository.MessageConsumeRecordRepository;
import com.shanks.framework.mq.service.MessageConsumeService;
import com.shanks.framework.mq.common.util.MQUtil;
import jakarta.annotation.Resource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.dao.DataIntegrityViolationException;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.Map;

@Service
public class MessageConsumeServiceImpl implements MessageConsumeService {

    private final Logger logger = LoggerFactory.getLogger(getClass());

    @Value("${rocketmq.namesrvAddr}")
    private String basicZoneNamesrvAddr;

    @Value("${rocketmq.grayNamesrvAddr:null}")
    private String canaryZoneNamesrvAddr;

    @Resource
    private MessageConsumeRecordRepository messageConsumeRecordRepository;

    @Resource
    private MessageConsumeExceptionRecordRepository messageConsumeExceptionRecordRepository;

    @Override
    public Boolean messageHasConsume(MQMessage message) {
        Boolean consume = messageConsumeRecordRepository.selectByMessage(message);
        if (!consume) {
            // 通过校验码再查询一次防止重复消费
            String checkCode = MQUtil.getCheckCode(message);
            consume = messageConsumeRecordRepository.selectByCheckCode(checkCode);
            if (consume)
                this.logger.info(" 消息已经被处理过, 忽略处理, 存在相同内容的消息：topic={}, tag={}, key={}, checkCode={} ", new Object[]{message
                        .getTopic(), message.getTags(), message.getKey(), checkCode});
        }
        return consume;
    }

    @Override
    public void deleteConsumeException(MQMessage message) {
        try {
            messageConsumeExceptionRecordRepository.deleteExceptionRecordByMsgTopicTagKey(message.getTopic(), message.getTags(), message.getKey());
        } catch (Exception e) {
            logger.error("删除消息消费异常记录失败", e);
            throw e;
        }
    }

    @Override
    public void addConsumeRecord(MQMessage mqMessage, String namesrvAddr, String consumerName) throws Exception {
        // 记录消息处理前的日志信息
        logger.info("开始记录消息消费：msgId={}, topic={}, tag={}, key={}",
                mqMessage.getMsgId(), mqMessage.getTopic(), mqMessage.getTags(), mqMessage.getKey());

        try {
            // 将 MQMessage 转换为 ConsumeRecord 并插入数据库
            MessageConsumeRecordPO messageConsumeRecordPO = convertToConsumeRecord(mqMessage, namesrvAddr, consumerName);
            messageConsumeRecordRepository.insert(messageConsumeRecordPO);

            // 记录成功插入的日志信息
            logger.info("成功记录消息消费：msgId={}, topic={}, tag={}, key={}",
                    mqMessage.getMsgId(), mqMessage.getTopic(), mqMessage.getTags(), mqMessage.getKey());
        } catch (DataIntegrityViolationException e) {
            // 处理重复消费的情况
            logger.error("检测到重复消费的消息：msgId={}, topic={}, tag={}, key={}. 错误详情: {}",
                    mqMessage.getMsgId(), mqMessage.getTopic(), mqMessage.getTags(), mqMessage.getKey(), e.getMessage());
            throw new MQDuplicatedConsumeException("检测到重复消费的消息", e);
        } catch (Exception e) {
            // 记录所有其他类型的异常，并抛出以供上层调用者处理
            logger.error("新增消息消费记录失败：msgId={}, topic={}, tag={}, key={}. 错误详情: {}",
                    mqMessage.getMsgId(), mqMessage.getTopic(), mqMessage.getTags(), mqMessage.getKey(), e.getMessage(), e);
            throw new Exception("新增消息消费记录失败", e);
        }
    }

    @Override
    public void addConsumeExceptionRecord(MQMessage mqMessage, String consumerName) throws Exception {
        try {
            logger.info("标记消息消费异常：msgId={}, topic={}, tag={}, key={} ", mqMessage
                    .getMsgId(), mqMessage.getTopic(), mqMessage.getTags(), mqMessage.getKey());
            messageConsumeExceptionRecordRepository.insertExceptionRecord(convertToConsumeExceptionRecord(mqMessage, consumerName));
        } catch (DuplicateKeyException e) {
            if (this.logger.isDebugEnabled())
                this.logger.debug("[{}] 消息重复消费，忽略处理. {}", mqMessage.getTopic() + "." + mqMessage.getTopic(),
                        LogUtil.formatAlert("MQ_WARN:REPEAT_CONSUME", "topicTag", mqMessage.getTopic() + "." + mqMessage.getTopic(), "msgKey", mqMessage.getKey()), e);
            throw e;
        } catch (Exception e) {
            this.logger.error("标记消息消费异常失败", e);
            throw e;
        }
    }

    private MessageConsumeRecordPO convertToConsumeRecord(MQMessage message, String zoneNamesrvAddr, String consumerName) throws Exception {
        MessageConsumeRecordPO record = new MessageConsumeRecordPO();
        record.setMsgId(message.getMsgId());
        record.setMsgTopic(message.getTopic());
        record.setMsgTag(message.getTags());
        record.setMsgKey(message.getKey());
        record.setConsumeTime(new Date());
        record.setCheckcode(MQUtil.getCheckCode(message));
        record.setMessageConsumeId(IdUtil.getSnowflakeNextId());
        record.setConsumerSource(consumerName);
        record.setMessageJson(message.getBodyAsString());
        Map<String, Object> props = message.getProperties();
        if (props != null) {
            String msgSource = (String) props.get(MQConstant.MSG.MSG_SOURCE);
            if (msgSource != null)
                record.setProducerSource(msgSource);
        }
        if (zoneNamesrvAddr != null)
            if (basicZoneNamesrvAddr.equals(zoneNamesrvAddr)) {
                record.setMqZone(MQZone.BASIC);
            } else if (canaryZoneNamesrvAddr.equals(zoneNamesrvAddr)) {
                record.setMqZone(MQZone.CANARY);
            }
        return record;
    }

    private MessageConsumeExceptionRecordPO convertToConsumeExceptionRecord(MQMessage message, String consumerName) throws Exception {
        MessageConsumeExceptionRecordPO record = new MessageConsumeExceptionRecordPO();
        record.setMsgId(message.getMsgId());
        record.setMsgTopic(message.getTopic());
        record.setMsgTag(message.getTags());
        record.setMsgKey(message.getKey());
        record.setConsumeExceptionTime(new Date());
        record.setMessageConsumeExceptionId(IdUtil.getSnowflakeNextId());
        record.setConsumer(consumerName);
        return record;
    }
}
